package spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StreamWordCount {
  def main(args: Array[String]): Unit = {

    // 1. 初始化Spark配置信息
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamWordCount")

    // 2. 初始化SparkStreamingContext
    val ssc: StreamingContext = new StreamingContext(conf = conf, batchDuration = Seconds(3))

    // 3. 通过监控端口创建DStream，读进来的数据为一行行
    val lineStreams: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    // 将每一行数据做切分，形成一个个单词 并将单词映射成元组（word,1）
    val wordStream: DStream[(String, Int)] = lineStreams.flatMap(_.split(" ")).map((_, 1))

    // 做词频统计
    val wordCountStream: DStream[(String, Int)] = wordStream.reduceByKey(_ + _)

    // print
    wordCountStream.print()

    // start SparkStreamContext and waiting for termination
    ssc.start()
    ssc.awaitTermination()
  }
}
